/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2024-2024. All rights reserved.
 * Description: hccl embedding service NPU
 * Author: baochangjin
 * Create: 2024-02-28
 */

#include <algorithm>
#include <list>
#include <vector>
#include <string>
#include <securec.h>
#include <hccl/hccl_types.h>

#include "hcom_private.h"
#include "config.h"
#include "externalinput_pub.h"
#include "externalinput.h"
#include "workflow_pub.h"
#include "hccl/base.h"
#include "hccl/hccl_ex.h"
#include "hccl/hcom_executor.h"
#include "runtime/dev.h"
#include "mem_device_pub.h"
#include "rank_consistentcy_checker.h"
#include "param_check_pub.h"
#include "adapter_hal.h"
#include "adapter_prof.h"
#include "dlra_function.h"
#include "dltdt_function.h"
#include "dlhdds_function.h"
#include "remote_access.h"
#include "alg_template_base_pub.h"
#include "topoinfo_detect.h"
#include "dlqos_function.h"
#include "profiling_manager.h"
#include "hccd_private.h"
#include "op_base.h"
#include "op_base_rpc.h"
#include "adapter_rts.h"
#include "device_capacity.h"
#include "topoinfo_ranktableParser_pub.h"
#include "hccl/hcom.h"
#include "mr_manager.h"

#include "hcom_rpc.h"
#include "hccl_comm_rpc.h"

#include "hcom_pub.h"
#include "op_base.h"
#include "transport_heterog_def.h"
#include "es_utils.h"

using namespace std;
using namespace hccl;

constexpr u32 THREE_TIMEGAP = 3;
extern constexpr u32 TIME_THREE_TIMEGAP = 3;

HcclResult HcpuGetCommByGroup(const char *group, std::shared_ptr<HcclCommBase> &hcclComm);

HcclResult HcclFinalizeCsComm(HcclComm comm)
{
    HcclUs startut = TIME_NOW();
    auto timeGap = TIME_NOW() - startut;
    u32 rankSize = 0;
    u32 rank = 0;

    // 入参检查
    CHK_PRT_RET(
        comm == nullptr, HCCL_WARNING("[Destroy][HcclHeterogComm]An empty comm given, skip destroy."), HCCL_SUCCESS);
    // 指针类型转换hcclComm*<-void*
    hccl::HcclCommBase* hcclComm = static_cast<hccl::HcclCommBase *>(comm);
    HCCL_RUN_INFO("Entry-HcclFinalizeCsComm: comm[%p].", comm);

    HcclResult ret = HCCL_SUCCESS;
    bool errorFlag = false;
    do {
        // 记录打印信息
        ret = hcclComm->GetUserRank(rank);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Finalize][HcclComm]errNo[0x%016llx] get user rank error.",
            HCCL_ERROR_CODE(ret)), errorFlag = true);
        ret = hcclComm->GetRankSize(rankSize);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Finalize][HcclComm]errNo[0x%016llx] get rank size error.",
            HCCL_ERROR_CODE(ret)), errorFlag = true);
    } while (0);

    std::string collectiveId = hcclComm->GetIdentifier();
    // 模型运行结束后hcom destroy时，将CheckInfo信息清空
    RankConsistentcyChecker::GetInstance().ClearCheckInfo();

    // 释放资源
    delete hcclComm;
    hcclComm = nullptr;

    if (errorFlag) {
        HCCL_ERROR("[Finalize][HcclComm]HcclFinalizeCsComm failed, return[0x%016llx].", HCCL_ERROR_CODE(ret));
        return ret;
    }

    HcclUs endut = TIME_NOW();
    /* 关键状态记录 */
    HCCL_RUN_INFO("HcclFinalizeCsComm success, take time [%lld]us, collectiveId[%s], rankSize[%u], rank[%u].",
        TAKE_TIME_US((endut - startut), (TIME_THREE_TIMEGAP * timeGap)), collectiveId.c_str(), rankSize, rank);
    return HCCL_SUCCESS;
}


HcclResult HcclRPCInitComm(const char* rankTableM, uint32_t rank, const CommAttr* attr, HcclComm* comm,
    HcomInfo &rankInfo)
{
    HcclResult ret = HCCL_SUCCESS;
    HcclUs startut = TIME_NOW();
    auto timeGap = TIME_NOW() - startut;

    // 入参合法性检查
    CHK_PTR_NULL(rankTableM);
    CHK_PTR_NULL(comm);
    CHK_PTR_NULL(attr);

    std::string rankTableStr = rankTableM;
    rankInfo.params.isHeterogComm = true;
    rankInfo.params.commWorkMode = attr->mode;

    /* 接口交互信息日志 */
    HCCL_RUN_INFO("Entry-HcclRPCInitComm:clusterInfo, rank[%u]", rank);
    if (attr->mode != WorkMode::HCCL_MODE_AI_CPU) {
        CHK_RET(DlRaFunction::GetInstance().DlRaFunctionInit());
    }

    /* --------------初始化------------------------- */
    bool errorFlag = false;
    hccl::HcclCommBase* pCommBase = nullptr;
    do {
        // 初始化外部参数
        ret = InitExternalInput();
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][HcclComm]errNo[0x%016llx] init external input error",
            HCCL_ERROR_CODE(ret)), errorFlag = true);

        ret = InitExternalInputHeterog();
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][HcclComm]errNo[0x%016llx] init external input error",
            HCCL_ERROR_CODE(ret)), errorFlag = true);

        // 解析rankTable_json对象，将解析的信息保存在rankinfo中，ranktableCRC计算
        ret = HcclParseRanktable(rankTableStr, to_string(rank), rankInfo.params, rankInfo.rankTable);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][HcclComm]errNo[0x%016llx] hccl analysis ranktable "\
            "info error:rank [%u]", HCCL_ERROR_CODE(ret), rank), errorFlag = true);

        // 生成通信域标识符
        ret = HcomGenerateCommId(rankInfo.params);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][OtherInfo]errNo[0x%016llx] generate CommId error",\
            HCCL_ERROR_CODE(HCCL_E_INTERNAL)), HCCL_E_INTERNAL);

        for (auto &rankIter : rankInfo.rankTable.rankList) {
            if (rankIter.rankId == rank) {
                if (attr->mode != WorkMode::HCCL_MODE_PS && attr->mode != WorkMode::HCCL_MODE_AI_CPU) {
                    rankIter.deviceInfo.devicePhyId = attr->deviceId;
                    break;
                }
            }
        }

        // new新对象
        pCommBase = reinterpret_cast<hccl::HcclCommBase *>
            (new (std::nothrow) hccl::HcclCommRpc(0, 0, rankInfo.rankTable.collectiveId));
        CHK_PTR_NULL(pCommBase);

        // 根据rankinfo结构体的内容,初始化通信域
        rankInfo.params.commHandle = pCommBase;
        rankInfo.params.attr = *attr;
        ret = reinterpret_cast<hccl::HcclCommRpc *>(pCommBase)->init(rankInfo.params, rankInfo.rankTable);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][HcclComm]errNo[0x%016llx] HeterogComm init error",
            HCCL_ERROR_CODE(ret)), errorFlag = true);

        // 打印rankTable信息
        ret = ShowRanktableConfigInfo(rankInfo.cloudFlag, rankInfo.params, rankInfo.rankTable);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][HcclComm]errNo[0x%016llx] put ranktable info error",
            HCCL_ERROR_CODE(ret)), errorFlag = true);

        *comm = pCommBase;
        rankInfo.psComm = pCommBase;
    } while (0);

    if (errorFlag) {
        HCCL_ERROR("[Init][CommClusterInfo]HeterogCommClusterInfto failed, return[0x%016llx]", HCCL_ERROR_CODE(ret));
        (void)HcclFinalizeCsComm(pCommBase);
        return ret;
    }

    HCCL_RUN_INFO("HcclRPCInitComm success,take time [%lld]us, collectiveId[%s], rankSize[%u], rank[%u]",
        TAKE_TIME_US((TIME_NOW() - startut), (THREE_TIMEGAP * timeGap)), pCommBase->GetIdentifier().c_str(),
        rankInfo.rankTable.rankNum, rank);
    return HCCL_SUCCESS;
}

HcclResult HcclPsAssociateWorkers(HcclComm comm, int32_t tag, uint32_t workerRanks[], uint64_t workerNum)
{
    CHK_PTR_NULL(comm);

    HcclUs startut = TIME_NOW();
    hccl::HcclCommBase* hcclComm = static_cast<hccl::HcclCommBase *>(comm);
    HCCL_RUN_INFO("Entry-%s: tag[%d], clientNum[%llu]", __func__, tag, workerNum);
    CHK_RET(hcclComm->PsAssociateWorkers(tag, workerRanks, workerNum));
    HCCL_RUN_INFO("HcclPsAssociateWorkers success,take time[%lld]us", DURATION_US(TIME_NOW() - startut));
    return HCCL_SUCCESS;
}

HcclResult HcclDestroyResouce(HcclComm comm, int32_t tag)
{
    CHK_PTR_NULL(comm);
    HcclUs startut = TIME_NOW();
    hccl::HcclCommBase* hcclComm = static_cast<hccl::HcclCommBase *>(comm);
    HCCL_RUN_INFO("Entry-HcclDestroyResouce started. tag[%d]", tag);
    CHK_RET(hcclComm->DestoryResource(tag));
    HCCL_RUN_INFO("HcclDestroyResouce success,take time[%lld]us", DURATION_US(TIME_NOW() - startut));
    return HCCL_SUCCESS;
}

HcclResult HcclInitCsComm(const char * rankTableM, s32 rankId, const char * roleTable,
    const CalcParams* calcParams, HcclComm* comm)
{
    CHK_PTR_NULL(rankTableM);
    CHK_PTR_NULL(roleTable);
    CHK_PTR_NULL(comm);
    HCCL_RUN_INFO("Entry-HcclInitCsComm, rankId[%d] rankTableM len[%u].", rankId, strlen(rankTableM));

    HcclUs startut = TIME_NOW();
    HcclResult ret = HCCL_SUCCESS;
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();

    if (!hcomInfo.isHcomInit) {
        HCCL_RUN_INFO("[Init][CsComm]Hcom is uninitialized, Entry HcclInitComm!!!");
        CommAttr attr;
        // host deviceId为-1，临时调测改为1
        attr.deviceId = 1;
        attr.mode = WorkMode::HCCL_MODE_PS;

        CHK_RET(HcclRPCInitComm(rankTableM, rankId, &attr, comm, hcomInfo));
    } else {
        HCCL_RUN_INFO("[Init][CsComm]Hcom is initialized!!!");
    }
    CHK_RET(DlHddsFunction::GetInstance().DlHddsFunctionInit());

    bool errorFlag = false;
    do {
        RoleTableInfo roleTableInfo;
        ret = CfgGetRoleTableInfo(roleTable, roleTableInfo);
        CHK_PRT_BREAK(ret != HCCL_SUCCESS,
            HCCL_ERROR("[Init][Result]errNo[0x%016llx] cfg get roletable[%s] info error:rankId[%d]",\
            HCOM_ERROR_CODE(ret), roleTable, rankId), errorFlag = true);

        if (!hcomInfo.isHcomInit) {
            ret = DlTdtFunction::GetInstance().DlTdtFunctionHeterogInit();
            CHK_PRT_BREAK(ret != HCCL_SUCCESS,
                HCCL_ERROR("[Init]errNo[0x%016llx] DlTdtFunctionHeterogInit error:rankId[%d]",
                HCOM_ERROR_CODE(ret), rankId), errorFlag = true);
            hccl::HcclCommRpc* hcclComm = static_cast<hccl::HcclCommRpc *>(*comm);
            ret = CheckRoleAndRankConsistent(roleTableInfo, hcomInfo.params, hcomInfo.rankTable);
            CHK_PRT_BREAK(ret != HCCL_SUCCESS,
                HCCL_ERROR("[Init]errNo[0x%016llx] CheckRoleAndRankConsistent info error:rankId[%d]",
                HCOM_ERROR_CODE(ret), rankId), errorFlag = true);
            ret = hcclComm->CsInit(roleTableInfo);
        } else {
            HcomInfo &hcomInfo = HcomGetCtxHomInfo();
            ret = CheckRoleAndRankConsistent(roleTableInfo, hcomInfo.params, hcomInfo.rankTable);
            CHK_PRT_BREAK(ret != HCCL_SUCCESS,
                HCCL_ERROR("[Init]errNo[0x%016llx] CheckRoleAndRankConsistent info error:rankId[%d]",
                HCOM_ERROR_CODE(ret), rankId), errorFlag = true);
            ret = hcomInfo.pCommBase->CsInit(roleTableInfo);
            CHK_PRT_BREAK(ret != HCCL_SUCCESS,
                HCCL_ERROR("[Init]errNo[0x%016llx] CsInit error:rankId[%d]",
                HCOM_ERROR_CODE(ret), rankId), errorFlag = true);
            if (!Is310PDevice()) {
                ret = LoadHRpcCsInitTask(rankTableM, rankId);
            }
        }
        CHK_PRT_BREAK(ret != HCCL_SUCCESS,
            HCCL_ERROR("[Init][Result]errNo[0x%016llx]hccl csComm init error", HCOM_ERROR_CODE(ret)),
            errorFlag = true);
    } while (0);

    if (errorFlag) {
        HCCL_ERROR("[Init][Result]hccl csComm init failed, rank[%u], return[%d]", hcomInfo.params.rank,
            HCOM_ERROR_CODE(ret));
        if (!hcomInfo.isHcomInit) {
            (void)HcclFinalizeCsComm(*comm);
        } else {
            (void)HcomDestroy();
        }

        return ret;
    }

    /* 关键状态记录 */
    HCCL_RUN_INFO("[HCCL_TRACE]hccl init cs comm success,take time [%lld]us, rankTableAddr[%p], rank[%d]",
        DURATION_US(TIME_NOW() - startut), rankTableM, rankId);

    return HCCL_SUCCESS;
}

HcclResult LoadHRpcCsInitTask(const char * rankTableM, s32 rankId)
{
    HCCL_RUN_INFO("Enter into LoadHRpcCsInitTask rankId[%d]", rankId);
    Stream stream(StreamType::STREAM_TYPE_ONLINE);
    rtKernelLaunchNames_t launchNames;
    launchNames.kernelName = "HdcsCsInit";
    launchNames.soName = "libccl_kernel.so";
    launchNames.opName = "CsInit";

    DeviceMem devicePid = DeviceMem::alloc(sizeof(s32));
    HdcsCsInitPara param{};
    param.psSize = 0;
    param.workerSize = 0;
    param.enableEntryLog = GetExternalInputHcclEnableEntryLog();
    std::map<u32, u32> psMap;
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    CHK_RET(hcomInfo.pCommBase->GetPsMap(psMap));
    for (auto it : psMap) {
        param.psList[it.first] = it.second;
        param.psSize++;
    }
    std::string temprRankTableM = rankTableM;
    s32 sRet = strncpy_s(param.rankTableM, HETEROG_RANK_TABLE_MAX_LEN, rankTableM, temprRankTableM.length());
    CHK_PRT_RET(sRet != EOK, HCCL_ERROR("[Init][CsComm] str copy fail. return[%d]", sRet), HCCL_E_INTERNAL);
    param.rank = rankId;
    param.devicePid = devicePid.ptr();

    CHK_RET(hrtAicpuKernelLaunch(launchNames, 1, &param, sizeof(HdcsCsInitPara), nullptr, stream.ptr()));
    CHK_RET(hcclStreamSynchronize(stream.ptr()));
    s32 devPid = 0;
    CHK_RET(hrtMemSyncCopy(&devPid, sizeof(s32), devicePid.ptr(), sizeof(s32),
        HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST));
    CHK_RET(hcomInfo.pCommBase->SetDevicePid(devPid));
    return HCCL_SUCCESS;
}

// 以进程粒度注册全局内存，多Server场景使用
HcclResult HcclRpcRegisterGlobalMemory(void* addr, u64 size)
{
    CHK_PTR_NULL(addr);
    HcclUs startut = TIME_NOW();
    HCCL_RUN_INFO("HcclRpcRegister Global addr[%llu] size[%llu]", std::hash<void *>{}(addr), size);
    HcclResult ret = MrManager::GetInstance().RegGlobalMr(addr, size);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[HcclRegisterGlobalMemory]errNo[0x%016llx] Hccl Register Global Memory failed, size[%llu].",
        HCCL_ERROR_CODE(ret), size), ret);
    HCCL_RUN_INFO("HcclRpcRegister Global success, take time [%lld]us.", DURATION_US(TIME_NOW() - startut));
    return HCCL_SUCCESS;
}

// 以进程粒度注销全局内存，多Server场景使用
HcclResult HcclRpcUnregisterGlobalMemory(void* addr)
{
    CHK_PTR_NULL(addr);
    HcclUs startut = TIME_NOW();
    HCCL_RUN_INFO("HcclRpcUnregister Global addr[%llu]", std::hash<void *>{}(addr));
    HcclResult ret = MrManager::GetInstance().DeRegGlobalMr(addr);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[HcclUnregisterGlobalMemory]errNo[0x%016llx] Hccl Unregister Global Memory failed.",
        HCCL_ERROR_CODE(ret)), ret);
    HCCL_RUN_INFO("HcclRpcUnregister Global success, take time [%lld]us.",
        DURATION_US(TIME_NOW() - startut));
    return HCCL_SUCCESS;
}

HcclResult HRpcCollRemotePairedParaCheck(const HcomRemoteOperationParams &params)
{
    CHK_PTR_NULL(params.keyAddr);
    CHK_PTR_NULL(params.value);
    CHK_PTR_NULL(params.tableId);
    CHK_PTR_NULL(params.indices);
    CHK_PTR_NULL(params.numUniqued);
    CHK_PTR_NULL(params.psSeg);
    CHK_PTR_NULL(params.psSegNum);

    return HCCL_SUCCESS;
}

HcclResult HcomRemoteLookup(s32 tag, s32 insertFlag, void *stream, s32 &intZerocpyFlag, s32 &outZerocpyFlag,
    RemoteLookupPubParams &lookupPubInfo, u64 keyNum)
{
    CHK_PTR_NULL(stream);
    CHK_PTR_NULL(lookupPubInfo.keys);
    CHK_PTR_NULL(lookupPubInfo.value);
    HcclUs startut = TIME_NOW();
    s32 streamId = 0;
    CHK_RET(hrtGetStreamId(stream, streamId));
    HCCL_RUN_INFO("Entry-HcomRemoteLookup, tag[%d] key max num[%llu] value item size[%d] keyNum[%llu], streamId[%d]"
        "intZerocpyFlag[%d] outZerocpyFlag[%d] max value item[%d]", tag, lookupPubInfo.keyMaxNum,
        lookupPubInfo.valueItemSize, keyNum, streamId, intZerocpyFlag, outZerocpyFlag, lookupPubInfo.maxEmbeddingDim);

    std::shared_ptr<hccl::HcclCommBase> hcclComm = nullptr;
    CHK_RET(HcpuGetCommByGroup(nullptr, hcclComm));
    u32 info = 0;

#ifndef CCL_LLT // 待板卡场景启用调用，代码合入时删除
    bool isBoardVersion{};
    CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->WorkerGetBoardVersion(isBoardVersion));
    CHK_RET(hrtDrvGetPlatformInfo(&info));
    if (isBoardVersion || info != 0) {
        HcomRemoteOperationParams params{};
        params.tableId = lookupPubInfo.tableId;
        params.keyAddr = lookupPubInfo.keys;
        params.value = lookupPubInfo.value;
        params.count = lookupPubInfo.keyMaxNum;
        params.keyNum = keyNum;
        params.valueDim = lookupPubInfo.valueItemSize / SIZE_TABLE[HCCL_DATA_TYPE_FP32];
        params.tag = tag;
        params.stream = stream;
        params.insertOption = insertFlag;
        params.flags = 0;
        params.intZerocpyFlag = intZerocpyFlag;
        params.outZerocpyFlag = outZerocpyFlag;
        params.disableUnique = true;
        params.maxEmbeddingDim = lookupPubInfo.maxEmbeddingDim;
        params.uniqueTag = lookupPubInfo.uniqueTag;
        CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->CollRemoteLookUp(params));
        HCCL_RUN_INFO("HcomRemoteLookup success tag[%d] take time[%lld]us", tag, DURATION_US(TIME_NOW() - startut));
        return HCCL_SUCCESS;
    }
#endif
    HcclRequest request = nullptr;
    CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->
        RemoteLookUp(tag, insertFlag, stream, intZerocpyFlag, outZerocpyFlag, lookupPubInfo, &request, keyNum));
    CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->
        WaitLookupData(tag, lookupPubInfo.value, stream, outZerocpyFlag, request));
    HCCL_RUN_INFO("HcomRemoteLookup success tag[%d] take time[%lld]us", tag, DURATION_US(TIME_NOW() - startut));
    return HCCL_SUCCESS;
}

HcclResult HcomCollRemoteUpdate(s32 tag, void* keys, u32 keyMaxNum, void *values, void *tableId, s32 insertOption,
    s32 valueItemSize, const char *group, void *stream, s32 maxEmbeddingDim, u64 keyNum)
{
    CHK_PTR_NULL(keys);
    CHK_PTR_NULL(values);
    CHK_PTR_NULL(tableId);
    CHK_PTR_NULL(stream);

    HcclUs startut = TIME_NOW();
    s32 streamId = 0;
    CHK_RET(hrtGetStreamId(stream, streamId));
    HCCL_RUN_INFO("Entry-HcomCollRemoteUpdate, tag[%d] key max num[%u]"
        "value item size[%d] keyNum[%llu] streamId[%d] maxEmbeddingDim[%d]",
        tag, keyMaxNum, valueItemSize, keyNum, streamId, maxEmbeddingDim);
    std::shared_ptr<hccl::HcclCommBase> hcclComm = nullptr;
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));

    CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->
        CollRemoteUpdate(tag, keys, keyMaxNum, values, tableId, insertOption, valueItemSize,
        stream, maxEmbeddingDim, keyNum));
    HCCL_RUN_INFO("HcomCollRemoteUpdate success.tag[%d] take time[%lld]us",
        tag, DURATION_US(TIME_NOW() - startut));
    return HCCL_SUCCESS;
}

HcclResult HcomCollRemoteLookup(s32 tag, void* keys, u32 keyMaxNum, void *values, void *tableId, s32 insertFlag,
    s32 valueItemSize, const char *group, void *stream, s32 flags, s32 maxEmbeddingDim, u64 keyNum)
{
    CHK_PTR_NULL(keys);
    CHK_PTR_NULL(values);
    CHK_PTR_NULL(tableId);
    CHK_PTR_NULL(stream);

    HcclUs startut = TIME_NOW();
    s32 streamId = 0;
    CHK_RET(hrtGetStreamId(stream, streamId));
    HCCL_RUN_INFO("Entry-HcomCollRemoteLookup, tag[%d] keyMaxNum[%u] valueItemSize[%d] flags[%d] keyNum[%llu] "
        "streamId[%d] max value dim size[%d]", tag, keyMaxNum, valueItemSize, flags, keyNum, streamId, maxEmbeddingDim);
    std::shared_ptr<hccl::HcclCommBase> hcclComm = nullptr;
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));

    CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->
        CollRemoteLookup(tag, keys, keyMaxNum, values, tableId, insertFlag,
        valueItemSize, stream, flags, maxEmbeddingDim, keyNum));
    HCCL_RUN_INFO("HcomCollRemoteLookup success.tag[%d] take time[%lld]us", tag, DURATION_US(TIME_NOW() - startut));
    return HCCL_SUCCESS;
}

HcclResult HcomCollRemoteLookupPaired(const HcomRemoteOperationParams &params)
{
    HcclUs startut = TIME_NOW();
    CHK_RET(HRpcCollRemotePairedParaCheck(params));
    CHK_PTR_NULL(params.stream);

    s32 streamId = 0;
    CHK_RET(hrtGetStreamId(params.stream, streamId));
    HCCL_RUN_INFO("Entry-HcomCollRemoteLookupPaired, tag[%d] key count[%d]"
        "value dim size[%d] streamId[%d] maxEmbeddingDim[%d]",
        params.tag, params.count, params.valueDim, streamId, params.maxEmbeddingDim);

    shared_ptr<hccl::HcclCommBase> hcclComm = nullptr;
    const char *group = params.group.empty() ? nullptr : params.group.c_str();
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));

    CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->CollRemoteLookupPaired(params));
    HCCL_RUN_INFO("HcomCollRemoteLookupPaired success take time[%lld]us tag[%d]",
        DURATION_US(TIME_NOW() - startut), params.tag);

    return HCCL_SUCCESS;
}

HcclResult HcomCollRemoteLookupUniquedAndPaired(const HcomRemoteOperationParams &params)
{
    HcclUs startut = TIME_NOW();
    CHK_RET(HRpcCollRemotePairedParaCheck(params));

    CHK_PTR_NULL(params.keyNumInput);
    CHK_PTR_NULL(params.uniqueIndices);
    CHK_PTR_NULL(params.stream);

    s32 streamId = 0;
    CHK_RET(hrtGetStreamId(params.stream, streamId));
    HCCL_RUN_INFO("Entry-HcomCollRemoteLookupUniquedAndPaired, tag[%d]"
        "key count[%d] value dim size[%d] streamId[%d] max value dim size[%d]",
        params.tag, params.count, params.valueDim, streamId, params.maxEmbeddingDim);

    shared_ptr<hccl::HcclCommBase> hcclComm = nullptr;
    const char *group = params.group.empty() ? nullptr : params.group.c_str();
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));

    CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->CollRemoteLookupUniquedPaired(params));
    HCCL_RUN_INFO("HcomCollRemoteLookupUniquedAndPaired success. take time[%lld]us tag[%d]",
        DURATION_US(TIME_NOW() - startut), params.tag);

    return HCCL_SUCCESS;
}

HcclResult HcomCollRemoteUpdatePaired(const HcomRemoteOperationParams &params)
{
    HcclUs startut = TIME_NOW();
    CHK_RET(HRpcCollRemotePairedParaCheck(params));

    CHK_PTR_NULL(params.stream);

    s32 streamId = 0;
    CHK_RET(hrtGetStreamId(params.stream, streamId));
    HCCL_RUN_INFO("Entry-HcomCollRemoteUpdatePaired, tag[%d] key count[%d]"
        "value dim size[%d] streamId[%d] max value dim size[%d]",
        params.tag, params.count, params.valueDim, streamId, params.maxEmbeddingDim);

    shared_ptr<hccl::HcclCommBase> hcclComm = nullptr;
    const char *group = params.group.empty() ? nullptr : params.group.c_str();
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));

    CHK_RET((reinterpret_cast<hccl::HcclCommRpc *>(hcclComm.get()))->CollRemoteUpdatePaired(params));
    HCCL_RUN_INFO("HcomCollRemoteUpdatePaired success take time[%lld]us tag[%d]",
        DURATION_US(TIME_NOW() - startut), params.tag);

    return HCCL_SUCCESS;
}

HcclResult HcclIsetLookupResponse(void* values, int32_t count, HcclDataType type, ServiceHandle handle,
    HcclComm comm, HcclRequest *request)
{
    CHK_PTR_NULL(values);
    CHK_PTR_NULL(handle);
    CHK_PTR_NULL(comm);
    CHK_PTR_NULL(request);
    HcclUs startut = TIME_NOW();
    HcclOperator<EmbeddingServiceParam> *paraPtr = static_cast<HcclOperator<EmbeddingServiceParam> *>(handle);
    hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(comm);
    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("Entry-HcclIsetLookupResponse, tag[%d] tableId[%u] values[%p] count[%d] HcclDataType[%s]",
            paraPtr->param_.tag, paraPtr->param_.tableId, values, count, GetDataTypeEnumStr(type).c_str());
    }

    bool isBoardVersion;
    CHK_RET(hcclComm->PsGetBoardVersion(isBoardVersion));
    bool isBoardOrGeneralServe = false;

    if (isBoardVersion || !IsGeneralServer()) {
        // 910*、多server非通用服务器场景
        isBoardOrGeneralServe = true;
        uint64_t beginTime = hrtMsprofSysCycleTime();
        CHK_RET(hcclComm->IsendLookupResponse(values, count, type, handle, request));
        auto &profilingManager = ProfilingManager::Instance();
        CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_ISEND_LOOKUP_RESPONSE));
    } else {
        // 310P 2PG soc
        CHK_RET(hcclComm->IsetLookupResponse(values, count, type, handle, request));
    }
    HcclRequestInfo* hcclReq = static_cast<HcclRequestInfo *>(*request);
    hcclReq->commHandle = hcclComm;

    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("HcclIsetLookupResponse success, tag[%d] tableId[%u] count[%d] HcclDataType[%s]"
            "isBoard[%u] take time[%lld]us", paraPtr->param_.tag, paraPtr->param_.tableId,
            count, GetDataTypeEnumStr(type).c_str(), isBoardOrGeneralServe, DURATION_US(TIME_NOW() - startut));
    }
    return HCCL_SUCCESS;
}

void LookupReqStatusToReqStatus(ReqStatus &status, LookupReqStatus &lStatus)
{
    status.tableId = lStatus.tableId;
    status.tag = lStatus.tag;
    status.actualSize = lStatus.actualCount;
    status.rsvd0 = lStatus.rsvd0;
}

HcclResult HcclGetLookupRequest(void* keys, int32_t count, HcclDataType type, s32 tag,
    ServiceHandle* handle, HcclComm comm, ReqStatus* status)
{
    CHK_PTR_NULL(comm);
    CHK_PTR_NULL(handle);
    CHK_PTR_NULL(keys);
    CHK_PTR_NULL(status);
    HcclUs startut = TIME_NOW();
    hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(comm);
    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("Entry-HcclGetLookupRequest, count[%d] data type[%s] tag[%d]", count,
            GetDataTypeEnumStr(type).c_str(), tag);
    }

    bool isBoardVersion;
    CHK_RET(hcclComm->PsGetBoardVersion(isBoardVersion));
    if (isBoardVersion || !IsGeneralServer()) {
        // 910*、多server非通用服务器场景
        uint64_t beginTime = hrtMsprofSysCycleTime();
        LookupReqStatus tmpLookupReqStatus{};
        do {
            CHK_RET(hcclComm->CollRecvLookupRequest(keys, count, type, tag, handle, &tmpLookupReqStatus, true));
        } while (tmpLookupReqStatus.actualCount == 0);

        HCCL_RUN_INFO("CollRecvLookupRequest success, data type[%s] tag[%d] inputCount[%d] "
            "actualCount[%d], tableId[%d] take time [%lld]us", GetDataTypeEnumStr(type).c_str(), tag, count,
            tmpLookupReqStatus.actualCount, tmpLookupReqStatus.tableId, DURATION_US(TIME_NOW() - startut));
        auto &profilingManager = ProfilingManager::Instance();
        CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime,
            ProfTaskType::TASK_COLL_RECV_LOOKUP_REQUEST));
        LookupReqStatusToReqStatus(*status, tmpLookupReqStatus);
        return HCCL_SUCCESS;
    }
    // 310P 2PG soc
    u64* key = reinterpret_cast<u64 *> (keys);
    CHK_RET(hcclComm->GetLookupRequest(tag, handle, key, count, status));

    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("HcclGetLookupRequest success, count[%d] data type[%s] tag[%d] take time[%lld]us",
            count, GetDataTypeEnumStr(type).c_str(), tag, DURATION_US(TIME_NOW() - startut));
    }
    return HCCL_SUCCESS;
}

HcclResult HddsCollRecvUpdateRequest(void* keys, int32_t keyCount, HcclDataType keyType, void* values,
    int32_t valueCount, HcclDataType valueType, int32_t tag,  ServiceHandle* handle, HcclComm comm,
    UpdateReqStatus* status)
{
    CHK_PTR_NULL(comm);
    CHK_PTR_NULL(handle);
    CHK_PTR_NULL(keys);
    CHK_PTR_NULL(status);
    CHK_PTR_NULL(values);

    HcclUs startut = TIME_NOW();
    hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(comm);
    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("Entry-HddsCollRecvUpdateRequest start, count[%d] data type[%s] tag[%d]", keyCount,
            GetDataTypeEnumStr(keyType).c_str(), tag);
    }
    uint64_t beginTime = hrtMsprofSysCycleTime();
    u64* key = reinterpret_cast<u64 *> (keys);

    TIME_PRINT(
        do { \
            CHK_RET(hcclComm->CollRecvUpdateRequest(key, keyCount, keyType, values, \
                valueCount, valueType, tag, handle, status)); \
        } while (status->actualKeyCount == 0);
        );

#ifdef ES_DATA_DFX
    u32 valueDim = status->actualValueCount / status->actualKeyCount;
    string printInfo = "ps update recv";
    EsUtils::PrintEsKvDataSummary(printInfo, static_cast<float *>(values), status->actualKeyCount, valueDim,
        static_cast<s64 *>(keys));
#endif

    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("HddsCollRecvUpdateRequest success, count[%d] data type[%s] tag[%d] inputKeyCount[%d] " \
            "inputValueCount[%d] actualKeyCount[%d] actualValueCount[%d], tableId[%d], " \
            "globalStep[%lld] take time[%lld]us", keyCount, GetDataTypeEnumStr(keyType).c_str(), tag, keyCount,
            valueCount, status->actualKeyCount, status->actualValueCount, status->tableId, status->globalStep,
            DURATION_US(TIME_NOW() - startut));
    }

    auto &profilingManager = ProfilingManager::Instance();
    CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_COLL_RECV_UPDATE_REQUEST));

    return HCCL_SUCCESS;
}

HcclResult HddsIsendUpdateResponse(ServiceHandle handle, HcclComm comm, HcclRequest* request)
{
    CHK_PTR_NULL(handle);
    CHK_PTR_NULL(comm);
    CHK_PTR_NULL(request);

    HcclUs startut = TIME_NOW();
    HcclOperator<EmbeddingServiceParam> *paraPtr = static_cast<HcclOperator<EmbeddingServiceParam> *>(handle);
    hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(comm);
    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("Entry-HddsIsendUpdateResponse tag[%d] tableId[%u]", paraPtr->param_.tag,
            paraPtr->param_.tableId);
    }

    uint64_t beginTime = hrtMsprofSysCycleTime();
    CHK_RET(hcclComm->IsendUpdateResponse(handle, request));
    HcclRequestInfo* hcclReq = static_cast<HcclRequestInfo *>(*request);
    hcclReq->commHandle = hcclComm;
    auto &profilingManager = ProfilingManager::Instance();
    CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_ISEND_UPDATE_RESPONSE));

    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("HddsIsendUpdateResponse success take time[%lld]us tag[%d] tableId[%u]",
            DURATION_US(TIME_NOW() - startut), paraPtr->param_.tag, paraPtr->param_.tableId);
    }

    return HCCL_SUCCESS;
}

HcclResult HddsCollRecvLookupRequest(void* keys, int32_t count, HcclDataType type, int32_t tag,
    ServiceHandle* handle, HcclComm comm, LookupReqStatus* status)
{
    CHK_PTR_NULL(comm);
    CHK_PTR_NULL(handle);
    CHK_PTR_NULL(keys);
    CHK_PTR_NULL(status);

    HcclUs startut = TIME_NOW();
    hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(comm);
    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("Entry-HddsCollRecvLookupRequest start, count[%d] data type[%s] tag[%d]", count,
            GetDataTypeEnumStr(type).c_str(), tag);
    }

    uint64_t beginTime = hrtMsprofSysCycleTime();

    TIME_PRINT(
        do { \
            CHK_RET(hcclComm->CollRecvLookupRequest(keys, count, type, tag, handle, status)); \
        } while (status->actualCount == 0);
    );

    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("HddsCollRecvLookupRequest success, data type[%s] tag[%d] inputCount[%d] actualCount[%d]"
            "tabId[%d] take time[%lld]us", GetDataTypeEnumStr(type).c_str(), tag, count, status->actualCount,
            status->tableId, DURATION_US(TIME_NOW() - startut));
    }

    auto &profilingManager = ProfilingManager::Instance();
    CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_COLL_RECV_LOOKUP_REQUEST));

    return HCCL_SUCCESS;
}

HcclResult HddsIsendLookupResponse(void* values, s32 count, HcclDataType type, ServiceHandle handle,
    HcclComm comm, HcclRequest *request)
{
    CHK_PTR_NULL(values);
    CHK_PTR_NULL(handle);
    CHK_PTR_NULL(comm);
    CHK_PTR_NULL(request);

    HcclUs startut = TIME_NOW();
    HcclOperator<EmbeddingServiceParam> *paraPtr = static_cast<HcclOperator<EmbeddingServiceParam> *>(handle);
    hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(comm);
    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("Entry-HddsIsendLookupResponse, count[%d] HcclDataType[%s] tag[%d] tableId[%u]",
            count, GetDataTypeEnumStr(type).c_str(), paraPtr->param_.tag, paraPtr->param_.tableId);
    }

    uint64_t beginTime = hrtMsprofSysCycleTime();
    CHK_RET(hcclComm->IsendLookupResponse(values, count, type, handle, request));
    HcclRequestInfo* hcclReq = static_cast<HcclRequestInfo *>(*request);
    hcclReq->commHandle = hcclComm;

    auto &profilingManager = ProfilingManager::Instance();
    CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_ISEND_LOOKUP_RESPONSE));

    if (GetExternalInputHcclEnableEntryLog()) {
        HCCL_RUN_INFO("HddsIsendLookupResponse success, count[%d] HcclDataType[%s] take time[%lld]us " \
            "tag[%d] tableId[%u]", count, GetDataTypeEnumStr(type).c_str(), DURATION_US(TIME_NOW() - startut),
            paraPtr->param_.tag, paraPtr->param_.tableId);
    }
    return HCCL_SUCCESS;
}

HcclResult HcclWaitSomeWithLog(int32_t count, HcclRequest requestArray[], s32* compCount, s32 compIndices[],
    HcclStatus compStatus[], bool needRunLog)
{
    CHK_PTR_NULL(compCount);
    CHK_PTR_NULL(requestArray);
    CHK_PTR_NULL(compIndices);
    CHK_PTR_NULL(compStatus);

    uint64_t beginTime = hrtMsprofSysCycleTime();
    auto &profilingManager = ProfilingManager::Instance();

    *compCount = 0;
    for (int i = 0; i < count; ++i) {
        HcclRequestInfo *hcclReq = reinterpret_cast<HcclRequestInfo *>(requestArray[i]);
        if (hcclReq == nullptr) {
            HCCL_WARNING("[%d]th hcclRequest is nullptr, no need to testSome", i);
            CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_WAIT_SOME));
            beginTime = hrtMsprofSysCycleTime();
            continue;
        }

        hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(hcclReq->commHandle);
        CHK_PTR_NULL(hcclComm);
        if (needRunLog && GetExternalInputHcclEnableEntryLog()) {
            HCCL_RUN_INFO("Entry-HcclWaitSome, count[%d]", count);
        }

        s32 completeCount = 0;
        s32 index = 0;
        TIME_PRINT(CHK_RET(hcclComm->WaitSome(1, &requestArray[i], &completeCount, &index, &compStatus[*compCount])));

        requestArray[i] = nullptr;
        compIndices[*compCount] = i;
        (*compCount)++;

        CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_WAIT_SOME));
        beginTime = hrtMsprofSysCycleTime();

        if (needRunLog && GetExternalInputHcclEnableEntryLog()) {
            HCCL_RUN_INFO("HcclWait : array[%d/%d] request[%p] comm[%s] peerRank[%u] tag[%d] type[%u]"
            "compCount[%d] status[%d]", i + 1, count, hcclReq,
            hcclComm->GetIdentifier().c_str(), hcclReq->transportRequest.epParam.src.rank,
            hcclReq->transportRequest.epParam.src.tag, hcclReq->transportRequest.requestType, *compCount,
            hcclReq->transportRequest.status);
        }

        // 目前是等待到一个完成就退出
        break;
    }

    return HCCL_SUCCESS;
}

HcclResult HcclWaitSome(int32_t count, HcclRequest requestArray[], s32* compCount, s32 compIndices[],
    HcclStatus compStatus[])
{
    // 需要打印Log
    CHK_RET(HcclWaitSomeWithLog(count, requestArray, compCount, compIndices, compStatus, true));

    return HCCL_SUCCESS;
}
HcclResult HcclAbortSelf(HcclComm comm, int32_t tag)
{
    CHK_PTR_NULL(comm);

    HCCL_RUN_INFO("Entry-HcclAbortSelf tag[%d]", tag);
    hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(comm);
    CHK_RET(hcclComm->AbortSelf(tag));

    HCCL_RUN_INFO("HcclAbortSelf success");
    return HCCL_SUCCESS;
}

HcclResult RegTransportLinks(s32 linkNum, void *transportPara, HcclComm comm)
{
    CHK_PTR_NULL(transportPara);
    CHK_PTR_NULL(comm);

    HCCL_RUN_INFO("Entry-RegTransportLinks");

    hccl::HcclCommRpc* hcclComm = reinterpret_cast<hccl::HcclCommRpc *>(comm);
    CHK_RET(hcclComm->RegTransportLinks(linkNum, transportPara));

    HCCL_RUN_INFO("RegTransportLinks success");
    return HCCL_SUCCESS;
}

HcclResult HddsServiceCancel(ServiceHandle handle)
{
    CHK_PTR_NULL(handle);
    auto opPtr = static_cast<HcclOperator<EmbeddingServiceParam> *>(handle);
    CHK_PTR_NULL(opPtr->param_.getRequestParam.commHandle);

    HCCL_RUN_INFO("Entry-HddsServiceCancel tag[%d]", opPtr->param_.tag);
    auto hcclComm = static_cast<hccl::HcclCommRpc *>(opPtr->param_.getRequestParam.commHandle);
    CHK_RET(hcclComm->ServiceCancel(handle));
    HCCL_RUN_INFO("HddsServiceCancel success");
    return HCCL_SUCCESS;
}

int HcclRPCImprobe(int srcRank, int tag, HcclComm comm, int* flag, HcclMessage* msg, HcclStatus* status)
{
    // 入参校验
    CHK_PTR_NULL(comm);
    CHK_PTR_NULL(flag);
    CHK_PTR_NULL(msg);
    CHK_PTR_NULL(status);

    // 关键日志记录
    hccl::HcclCommRpc* hcclComm = static_cast<hccl::HcclCommRpc *>(comm);
    u32 localRank = INVALID_VALUE_RANKID;
    CHK_RET(hcclComm->GetUserRank(localRank));
    HCCL_INFO("Entry-HcclRPCImprobe: comm[%s] srcRank[%d] localRank[%u] tag[%d]",
        hcclComm->GetIdentifier().c_str(), srcRank, localRank, tag);
    // 获取 recv request entry
    CHK_RET(hcclComm->Improbe(static_cast<u32>(srcRank), tag, *flag,
        *msg, *status));
    HcclMessageInfo* hcclMsg = static_cast<HcclMessageInfo *>(*msg);
    if (*flag == HCCL_IMPROBE_COMPLETED) {
        hcclMsg->commHandle = comm;
    }
    // 关键日志记录
    HCCL_INFO("HcclRPCImprobe success. comm[%s] srcRank[%d] localRank[%u] tag[%d]: flag[%d] status[%d] count[%d] "
        "msg[%p].", hcclComm->GetIdentifier().c_str(), srcRank, localRank, tag, *flag, status->error, status->count,
        *msg);
    return HCCL_SUCCESS;
}

int HcclRPCGetCount(const HcclStatus *status, HcclDataType dataType, int *count)
{
       // 入参校验
    CHK_PTR_NULL(status);
    CHK_PTR_NULL(count);

    if (status->error != 0) {
        HCCL_WARNING("HcclRPCGetCount::Failed to obtain the count status[%d].", status->error);
        return HCCL_E_PARA;
    }

    *count = status->count;

     // 关键日志记录
    HCCL_INFO("HcclRPCGetCount success. peerRank[%d] tag[%d] status[%d] dataType[%s] count[%d].",
        status->srcRank, status->tag, status->error, GetDataTypeEnumStr(dataType).c_str(), *count);
    return HCCL_SUCCESS;
}

int HcclRPCImrecv(void* buffer, int count, HcclDataType dataType, HcclMessage* msg, HcclRequest* request)
{
    CHK_PTR_NULL(msg);
    CHK_PTR_NULL(*msg);
    CHK_PTR_NULL(request);

    HcclMessageInfo* hcclMsg = static_cast<HcclMessageInfo *>(*msg);
    CHK_PTR_NULL(hcclMsg->commHandle);
    hccl::HcclCommRpc* hcclComm = static_cast<hccl::HcclCommRpc *>(hcclMsg->commHandle);
    uint32_t peerRank = hcclMsg->envelope.envelope.epParam.src.rank;
    uint32_t tag = hcclMsg->envelope.envelope.epParam.src.tag;
    /* 接口交互信息日志 */
    HCCL_INFO("Entry-HcclRPCImrecv: comm[%s] peerRank[%u] tag[%u]: addr[%p] count[%d] dtype[%s] msg[%p]",
        hcclComm->GetIdentifier().c_str(), peerRank, tag, buffer, count, GetDataTypeEnumStr(dataType).c_str(), *msg);

    CHK_RET(hcclComm->Imrecv(buffer, count, dataType, *msg, *request));

    HcclRequestInfo* hcclReq = static_cast<HcclRequestInfo *>(*request);
    hcclReq->commHandle = hcclComm;
    /* 关键状态记录 */
    HCCL_INFO("HcclRPCImrecv success. comm[%s] peerRank[%u] tag[%u]: addr[%p] count[%d] dtype[%s] msg[%p] request[%p].",
        hcclComm->GetIdentifier().c_str(), peerRank, tag, buffer, count, GetDataTypeEnumStr(dataType).c_str(), *msg,
        *request);
    return HCCL_SUCCESS;
}

HcclResult HcomRpcGetDeviceType(const char *group, DevType &deviceType)
{
    shared_ptr<hccl::HcclCommBase> hcclComm = nullptr;
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));

    CHK_RET(hcclComm->GetDevType(deviceType));

    return HCCL_SUCCESS;
}

HcclResult HcomRpcGetHccsLinkNum(const char *group, u32 &numHccsLink)
{
    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    CHK_RET(HcpuGetCommByGroup(group, hcclComm));
    CHK_RET(hcclComm->GetHccsLinkNum(numHccsLink));
    return HCCL_SUCCESS;
}

HcclResult HcomRpcGetRankSize(const char *group, u32 *rankSize)
{
    RPT_INPUT_ERR(rankSize == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "parameter", "value", "tips"}),\
        std::vector<std::string>({"HcomGetRankSize", "rankSize", "nullptr", "please check rankSize"}));
    CHK_PTR_NULL(rankSize);
    bool &isAutoTuneModeOpen = HcomGetCtxAutoTuneMode();
    if (isAutoTuneModeOpen) {
        *rankSize = 1;
        return HCCL_SUCCESS;
    }

    HcclResult ret = HcomCheckGroupName(group);
    RPT_INPUT_ERR(ret != HCCL_SUCCESS, "EI0003", std::vector<std::string>({"ccl_op", "parameter", "value", "tips"}),
        std::vector<std::string>({
        "HcomGetRankSize",
        "group",
        {group, strnlen(group, GROUP_NAME_MAX_LEN + 1)},
        "please check group name"
        }));
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Get][RankSize]errNo[0x%016llx] group name is invalid", HCOM_ERROR_CODE(ret)), ret);

    std::shared_ptr<hccl::HcclCommBase> hcclComm;
    if (HcpuGetCommByGroup(group, hcclComm) == HCCL_SUCCESS) {
        CHK_RET(hcclComm->GetRankSize(*rankSize));
    } else {
        std::string strGroup = (group == nullptr) ? HCCL_WORLD_GROUP : group;
        ret = GetGroupRankInfo(strGroup.c_str(), RankInfoType::RANK_SIZE_IN_GROUP, 0, rankSize);
        CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Get][RankSize]errNo[0x%016llx] get group[%s] rank info error",
        HCOM_ERROR_CODE(ret), strGroup.c_str()), ret);
        HCCL_INFO("hcom rpc get rank size success, group[%s]", strGroup.c_str());
    }
    return HCCL_SUCCESS;
}

HcclResult HcomRpcGetInitStatus(bool& initiated)
{
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    initiated = !(hcomInfo.pCommBase == nullptr);

    HCCL_INFO("Get Hcom Init Status: [%d]", initiated);
    return HCCL_SUCCESS;
}

bool HcomIsRpcComm(const char *group)
{
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    return (hcomInfo.pCommBase != nullptr);
}

HcclResult HcomRpcGetRankId(const char *group, u32 *rankId)
{
    RPT_INPUT_ERR(rankId == nullptr, "EI0003", std::vector<std::string>({"ccl_op", "parameter", "value", "tips"}),\
        std::vector<std::string>({"HcomGetRankId", "rankId", "nullptr", "please check rankId"}));
    CHK_PTR_NULL(rankId);
    bool &isAutoTuneModeOpen = HcomGetCtxAutoTuneMode();
    if (isAutoTuneModeOpen) {
        *rankId = 0;
        return HCCL_SUCCESS;
    }
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    CHK_PRT_RET(hcomInfo.pCommBase == nullptr, HCCL_ERROR("[Get][RankId]hcomInfo.pCommBase is null, "\
        "please check if the initialize process is called."), HCCL_E_PTR);
    HcclResult ret = HcomCheckGroupName(group);
    RPT_INPUT_ERR(ret != HCCL_SUCCESS,
        "EI0003", std::vector<std::string>({ "ccl_op", "parameter", "value", "tips" }),
        std::vector<std::string>({
        "HcomGetRankId",
        "group",
        { group, strnlen(group, GROUP_NAME_MAX_LEN + 1) },
        "please check group name"
        }));
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Get][RankId]errNo[0x%016llx] get_rank_id group name is invalid", HCOM_ERROR_CODE(ret)), ret);

    std::string strGroup = (group == nullptr) ? HCCL_WORLD_GROUP : group;

    CHK_RET(GetGroupRankInfo(strGroup.c_str(), RankInfoType::RANK_ID_IN_GROUP, 0, rankId));

    HCCL_INFO("hcom get rank id success, group[%s]", strGroup.c_str());

    return HCCL_SUCCESS;
}

HcclResult HcomRpcUnloadTask(const std::string group, const char *tag)
{
    // UnloadTask和HcomDestroyGroup都会释放group资源, 避免重复释放
    HCCL_INFO("[UnloadTask]HcomRpcUnloadTask: tag[%s]", tag);
    HcomInfo &hcomInfo = HcomGetCtxHomInfo();
    std::string strGroup = (group.c_str() == nullptr) ? HCCL_WORLD_GROUP : group;
    if (strGroup != HCCL_WORLD_GROUP && hcomInfo.hcomGroupMap.find(group) == hcomInfo.hcomGroupMap.end()) {
        HCCL_WARNING("[UnloadTask]group[%s] is not exist, do nothing.", group.c_str());
        return HCCL_SUCCESS;
    }

    std::shared_ptr<HcclCommBase> hcclCommbase;
    CHK_RET(HcpuGetCommByGroup(group.c_str(), hcclCommbase));
    CHK_PRT_RET(hcclCommbase == nullptr, HCCL_WARNING("[UnloadAllTask]hcclComm is null, "\
    "please check if the initialize process is called."), HCCL_SUCCESS);
    CHK_RET(hcclCommbase->ClearOpResource(tag));

    return HCCL_SUCCESS;
}

HcclResult HcomRpcSetWorkspaceResource(const std::string &tag, const char *group, std::vector<rtStream_t> stream,
    void *memPtr, u64 maxSize)
{
    if (group == nullptr) {
        group = HCCL_WORLD_GROUP;
    }

    HCCL_RUN_INFO("Entry HcomRpcSetWorkspaceResource tag[%s] group[%s] memPtr[%llu] maxSize[%llu]",
        tag.c_str(), group, std::hash<void *>{}(memPtr), maxSize);
    std::shared_ptr<hccl::HcclCommBase> hcclCommcpu;
    CHK_RET(HcpuGetCommByGroup(group, hcclCommcpu));
    CHK_RET(hcclCommcpu->SetWorkspaceResource(tag, memPtr, maxSize, stream));

    return HCCL_SUCCESS;
}

HcclResult HcomRpcGetTopoDesc(const char *group, HcclTopoDescs *topoDescs, uint32_t topoSize)
{
    CHK_PTR_NULL(topoDescs);
    CHK_PTR_NULL(group);

    std::shared_ptr<HcclCommBase> hcclCommbase;
    CHK_RET(HcclDeviceRefresh());

    CHK_RET(HcpuGetCommByGroup(group, hcclCommbase));
    CHK_RET(hcclCommbase->GetTopoDesc(topoDescs, topoSize));

    return HCCL_SUCCESS;
}
